/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.cratos.xxgram.platf.message;

import java.net.InetSocketAddress;
import java.security.PrivateKey;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.logging.Level;
import java.util.stream.Collectors;

import javax.annotation.Resource;

import org.redkale.boot.Application;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.http.RestHeader;
import org.redkale.net.http.RestMapping;
import org.redkale.net.http.RestService;
import org.redkale.net.http.WebSocket;
import org.redkale.net.http.WebSocketNode;
import org.redkale.service.RetResult;
import org.redkale.source.ColumnValue;
import org.redkale.source.DataSource;
import org.redkale.source.FilterExpress;
import org.redkale.source.FilterFunc;
import org.redkale.source.FilterNode;
import org.redkale.source.Flipper;
import org.redkale.util.AnyValue;
import org.redkale.util.Comment;
import org.redkale.util.SelectColumn;
import org.redkale.util.Sheet;
import org.redkale.util.Utility;

import com.cratos.xxgram.platf.base.BaseService;
import com.cratos.xxgram.platf.base.RetCodes;
import com.cratos.xxgram.platf.base.UserInfo;
import com.cratos.xxgram.platf.file.FileService;
import com.cratos.xxgram.platf.friend.FriendService;
import com.cratos.xxgram.platf.friend.FriendShip;
import com.cratos.xxgram.platf.group.GroupDetail;
import com.cratos.xxgram.platf.group.UserGroup;
import com.cratos.xxgram.platf.message.BasicContent.SysNoticeMessage;
import com.cratos.xxgram.platf.message.MessageResult.SendMsgResponse;
import com.cratos.xxgram.platf.message.NotifyContent.GroupReadReceiptNotify;
import com.cratos.xxgram.platf.message.NotifyContent.SynMulTermalNotify;
import com.cratos.xxgram.platf.push.JpushService;
import com.cratos.xxgram.platf.sysnotice.SystemNotice;
import com.cratos.xxgram.platf.sysnotice.SystemUserNotice;
import com.cratos.xxgram.platf.user.UserDetail;
import com.cratos.xxgram.platf.user.UserService;
import com.cratos.xxgram.platf.util.Cryptos;

/**
 *
 * @author SquRab
 */
@RestService(name = "message", comment = "消息处理服务")
public class MessageService extends BaseService {

	@Resource
	protected JsonConvert jsonConvert = JsonConvert.root();

	@Resource
	protected FriendService friendService;

	@Resource
	protected FileService fileService;

	@Resource
	protected JpushService pushService;

	@Resource
	protected UserService userService;
	
	
	@Comment("定时任务")
	protected ScheduledThreadPoolExecutor scheduler;

	Cryptos.ECDHCrypto crypto = Cryptos.ECDHCrypto.getInstance();

	private final Map<Integer, String> filterMultilMsg = new HashMap<Integer, String>();
	@Comment("H5登录重联 session 缓存 key sessionid ,value =userid-time36")
	protected static final ConcurrentHashMap<String, String> h5SessionMapper = new ConcurrentHashMap<String, String>();

	private void initH5Connector(int threadnum) {
		this.scheduler = new ScheduledThreadPoolExecutor(threadnum + 2, (Runnable r) -> {
			final Thread t = new Thread(r, "messageService" + "-task-Thread");
			t.setDaemon(true);
			return t;
		});

		// 处理H5 每隔30s 遍历一下登录状态
		this.scheduler.scheduleAtFixedRate(() -> {
			try {
				long now = System.currentTimeMillis();
				logger.finest("web 端下线... 发送 app 状态消息 开始 === h5SessionMapper" + jsonConvert.convertTo(h5SessionMapper));
				h5SessionMapper.forEach((jsessionId, val) -> {
					String[] vals = val.split("-");
					int userid = Integer.parseInt(vals[0]);
					long time = Long.parseLong(vals[1], 36);
					// 超过30 s直接视为下线
					if (now - time > 1000 * 20) {
						
						sendDisableH5Msg(jsessionId, userid);
						h5SessionMapper.remove(jsessionId);
					}
				});
				logger.finest("web 端下线... 发送 app 状态消息完成 !");
			} catch (Throwable e) {
				logger.log(Level.SEVERE, "scheduleAtFixedRate sendDisableH5Msg error", e);
			}
		}, 1, 20, TimeUnit.SECONDS);
		this.scheduler.setRemoveOnCancelPolicy(true);
	}

	/**
	 * 动态加载类文件 消息类型 -> 消息内容对象
	 *
	 * @param config
	 */
	public void init(AnyValue config) {
		super.init(config); // To change body of generated methods, choose Tools | Templates.
		//initH5Connector(2);// h5连接个数处理
		filterMultilMsg.put(1001, "文本消息");
		filterMultilMsg.put(1002, "语音");
		filterMultilMsg.put(1003, "文件类的消息");
		filterMultilMsg.put(1004, "小视频消息");
		
		
		filterMultilMsg.put(1005, "位置消息");
		filterMultilMsg.put(1006, "图文消息");
		filterMultilMsg.put(1007, "群聊自定义消息");
		filterMultilMsg.put(1008, "文件类分享转发");
		
		
		filterMultilMsg.put(1009, "音频消息");
		filterMultilMsg.put(1010, "图片消息");
		filterMultilMsg.put(1012, "分享名片");
		filterMultilMsg.put(1013, "融云语音通话消息");
		filterMultilMsg.put((int)NotifyContent.OBJ_RECALL_MESSAGE_NTF, "消息撤回");
                filterMultilMsg.put((int)NotifyContent.OBJ_RECALLALL_MESSAGE_NTF, "消息全部");
		//
		// 统一保存
	}

	/**
	 * 群消息回执过滤
	 * @param record
	 * @param recUserids
	 */
	private void filterGroupRecepit(MessageRecord record, Set<Integer> recUserids) {
		if (record.getMsgtype() == GroupReadReceiptNotify.TYPE) {
			recUserids.clear();
			GroupReadReceiptNotify notify = (GroupReadReceiptNotify) record.getContentObj();
			// 只回执发送用户
			recUserids.addAll(notify.getMessageids().stream().map((messageId -> {return Integer.parseInt(messageId.split("-")[0]);})).collect(Collectors.toSet()));
		}
	}
	@Comment("handler 消息发送")
	public CompletableFuture<Integer> sendMessage(MessageRecord message) {
		CompletableFuture<Integer> result = new CompletableFuture<>();
		logger.finest("sendMessage message === " + jsonConvert.convertTo(message));
		Integer fromId = message.getFromid();
		Integer targetId = message.getTargetid();
		// 群聊
		if (message.getMsgkind() == MessageRecord.MSG_KIND_GROUP) {
			Collection<UserGroup> rows = source.queryList(UserGroup.class,
					FilterNode.create("groupid", FilterExpress.EQUAL, targetId));
			if (rows != null && rows.size() > 0) {
				Set<Integer> recUserids = rows.stream().filter(
						(obj -> fromId != obj.getUserid() || message.getMsgtype() == BasicContent.TipMessage.TYPE)) // 过滤掉发起用户
																													// tip
																													// 消息所有人都发送
						.map(o -> o.getUserid()).collect(Collectors.toSet());
				
				// 只回执发送用户的消息
				filterGroupRecepit(message, recUserids);
				recUserids.forEach(reciver -> {
					wsmessageQueue.add(() -> {
						// 发送失败或者 用户不在线就保存离线消息
						if (webSocketNode != null) {
							webSocketNode.sendMessage(Utility.ofMap("record", message), reciver)
									.whenComplete((r, e) -> {
										logger.finest("----sendMessage----websocket send status == " + r
												+ "===kind =====" + message.getMsgkind() + "execption ===" + e
												+ "--------------");
										// 发送失败或者App 不再先 都发送消息
										if (!validateSendStatus(r, e) || !appOnline(reciver)) {
											insertQueue.add(message.copy(reciver));// 添加离线消息
											try {
												pushService.pushMessage(message, Arrays.asList(reciver)); // 推送离线消息
												logger.finest("message send res ===" + jsonConvert.convertTo(r));
											} catch (Exception error) {
												logger.finest("message send error ===" + error.getMessage());
											}
											result.completeExceptionally(new RuntimeException("send fail ..."));
										}
									});
						} else {
							insertQueue.add(message.copy(reciver));// 添加离线消息
							try {
								RetResult r = pushService.pushMessage(message, recUserids); // 推送离线消息
								logger.finest("message send res ===" + jsonConvert.convertTo(r));
							} catch (Exception e) {
								logger.finest("message send error ===" + e.getMessage());
							}

						}
					});
				});
			}
		}
		// 私聊
		if (message.getMsgkind() == MessageRecord.MSG_KIND_PRIVATE) {
			wsmessageQueue.add(() -> {
				// 发送失败或者 用户不在线就保存离线消息
				if (webSocketNode != null) {
					webSocketNode.sendMessage(Utility.ofMap("record", message), targetId).whenComplete((r, e) -> {
						logger.finest("----sendMessage----websocket send status == " + r + "===kind ====="
								+ message.getMsgkind() + "execption ===" + e + "--------------");
						//发送成功或者App 不再先都发送消息
						if (!validateSendStatus(r, e)  || !appOnline(targetId)) {
							insertQueue.add(message.copy(targetId));// 添加离线消息
							try {
								pushService.pushMessage(message, Arrays.asList(targetId)); // 推送离线消息
								logger.finest("message send res ===" + jsonConvert.convertTo(r));
							} catch (Exception error) {
								logger.finest("message send error ===" + error.getMessage());
							}
							result.completeExceptionally(new RuntimeException("send fail ..."));
						}
					});
				} else {
					insertQueue.add(message.copy(targetId));// 添加离线消息
					try {
						RetResult r = pushService.pushMessage(message, Arrays.asList(targetId)); // 推送离线消息
						logger.finest("message send res ===" + jsonConvert.convertTo(r));
					} catch (Exception e) {
						logger.finest("message send error ===" + e.getMessage());
					}
				}
			});
		}
		// 系统消息
		if (message.getMsgkind() == MessageRecord.MSG_KIND_SYSTEM) {
			wsmessageQueue.add(() -> {

				// 发送失败或者 用户不在线就保存离线消息
				if (webSocketNode != null) {
					webSocketNode.sendMessage(Utility.ofMap("record", message), targetId).whenComplete((r, e) -> {
						logger.finest("----sendMessage----websocket send status == " + r + "===kind ====="
								+ message.getMsgkind() + "execption ===" + e + "--------------");
						if (!validateSendStatus(r, e)  || !appOnline(targetId)) {
							insertQueue.add(message.copy(targetId));// 添加离线消息
							try {
								pushService.pushMessage(message, Arrays.asList(targetId)); // 推送离线消息
								logger.finest("message send res ===" + jsonConvert.convertTo(r));
							} catch (Exception error) {
								logger.finest("message send error ===" + error.getMessage());
							}
							result.completeExceptionally(new RuntimeException("send fail ..."));
						}
					});
				} else {
					insertQueue.add(message.copy(targetId));// 添加离线消息
					try {
						RetResult r = pushService.pushMessage(message, Arrays.asList(targetId)); // 推送离线消息
						logger.finest("message send res ===" + jsonConvert.convertTo(r));
					} catch (Exception e) {
						logger.finest("message send error ===" + e.getMessage());
					}
				}
			});
		}
		return result;
	}

	@Comment("handler 給用戶發送消息 不在線進行保存")
	public CompletableFuture<Integer> sendNotifyMessageWithSave(MessageRecord message, Collection<Integer> userids) {
		CompletableFuture<Integer> result = new CompletableFuture<>();
		logger.finest("sendMessage message === " + jsonConvert.convertTo(message) + ", userids ===="
				+ jsonConvert.convertTo(userids));
		if (userids != null) {
			userids.forEach(receiver -> {
				wsmessageQueue.add(() -> {
					// 发送失败或者 用户不在线就保存离线消息
					if (webSocketNode != null) {
						webSocketNode.sendMessage(Utility.ofMap("record", message), receiver).whenComplete((r, e) -> {
							logger.finest("----sendNotifyMessageWithSave websocket send status = " + r + "execption ="
									+ e + ",data=" + jsonConvert.convertTo(message));
							if (!validateSendStatus(r, e) || !appOnline(receiver)) {
								insertQueue.add(message.copy(receiver));// 添加离线消息
								RetResult rr = null;
								try {
									rr = pushService.pushMessage(message, Arrays.asList(receiver)); // 推送离线消息
									logger.finest("message send res ===" + jsonConvert.convertTo(rr));
								} catch (Exception ex) {
									logger.finest("message send error ===" + jsonConvert.convertTo(rr)
											+ "== message ===" + ex.getMessage());
								}

								result.completeExceptionally(new RuntimeException("send fail ..."));
							}
						});
					} else {
						insertQueue.add(message.copy(receiver));// 添加离线消息
						RetResult rr = null;
						try {
							rr = pushService.pushMessage(message, Arrays.asList(receiver)); // 推送离线消息
							logger.finest("message send res ===" + jsonConvert.convertTo(rr));
						} catch (Exception ex) {
							logger.finest("message send error ===" + jsonConvert.convertTo(rr) + "== message ==="
									+ ex.getMessage());
						}
					}
				});
			});
		}

		return result;
	}

        @Comment("handler 給用戶發送消息 不在線進行保存")
	public CompletableFuture<Integer> sendNotifyMessageWithSaveOss(MessageRecord message, Collection<Integer> userids, boolean isOnline) {
		CompletableFuture<Integer> result = new CompletableFuture<>();
		logger.finest("sendMessage message === " + jsonConvert.convertTo(message) + ", userids ===="
				+ jsonConvert.convertTo(userids));
		if (userids != null) {
			userids.forEach(receiver -> {
				wsmessageQueue.add(() -> {
					// 发送失败或者 用户不在线就保存离线消息
					if (webSocketNode != null) {
						webSocketNode.sendMessage(Utility.ofMap("record", message), receiver).whenComplete((r, e) -> {
							logger.finest("----sendNotifyMessageWithSave websocket send status = " + r + "execption ="
									+ e + ",data=" + jsonConvert.convertTo(message));
							if (!validateSendStatus(r, e) || !isOnline) {
								insertQueue.add(message.copy(receiver));// 添加离线消息
								RetResult rr = null;
								try {
									rr = pushService.pushMessage(message, Arrays.asList(receiver)); // 推送离线消息
									logger.finest("message send res ===" + jsonConvert.convertTo(rr));
								} catch (Exception ex) {
									logger.finest("message send error ===" + jsonConvert.convertTo(rr)
											+ "== message ===" + ex.getMessage());
								}

								result.completeExceptionally(new RuntimeException("send fail ..."));
							}
						});
					} else {
						insertQueue.add(message.copy(receiver));// 添加离线消息
						RetResult rr = null;
						try {
							rr = pushService.pushMessage(message, Arrays.asList(receiver)); // 推送离线消息
							logger.finest("message send res ===" + jsonConvert.convertTo(rr));
						} catch (Exception ex) {
							logger.finest("message send error ===" + jsonConvert.convertTo(rr) + "== message ==="
									+ ex.getMessage());
						}
					}
				});
			});
		}

		return result;
	}
        
	@Comment("handler 消息发送 不保存")
	public CompletableFuture<Integer> sendNotifyMessage(MessageRecord message, List<Integer> receivers) {
		CompletableFuture<Integer> result = new CompletableFuture<>();
		logger.finest("sendMessage message === " + jsonConvert.convertTo(message));
		wsmessageQueue.add(() -> {
			if (webSocketNode == null) {
				logger.log(Level.FINEST, "sendMap: " + JsonConvert.root().convertTo(message));
				result.complete(0);
				return;
			}
			// 发送失败或者 用户不在线就保存离线消息
			webSocketNode.sendMessage(Utility.ofMap("record", message), receivers.stream()).whenComplete((r, e) -> {
				if (!validateSendStatus(r, e)) {
					RetResult rr = null;
					try {
						rr = pushService.pushMessage(message, receivers); // 推送离线消息
						logger.finest("message send res ===" + jsonConvert.convertTo(rr));
					} catch (Exception ex) {
						logger.finest("message send error ===" + jsonConvert.convertTo(rr) + "== message ==="
								+ ex.getMessage());
					}

					result.completeExceptionally(new RuntimeException("send fail ..."));
				}
			});
		});
		return result;
	}

        @Comment("只发送不推送不保存")
	public CompletableFuture<Integer> sendNotifyMessageOnly(MessageRecord message, List<Integer> receivers) {
		CompletableFuture<Integer> result = new CompletableFuture<>();
		logger.finest("sendMessage message === " + jsonConvert.convertTo(message));
		wsmessageQueue.add(() -> {
			if (webSocketNode == null) {
				logger.log(Level.FINEST, "sendMap: " + JsonConvert.root().convertTo(message));
				result.complete(0);
				return;
			}
			// 发送失败或者 用户不在线就保存离线消息
			webSocketNode.sendMessage(Utility.ofMap("record", message), receivers.stream()).whenComplete((r, e) -> {
				if (!validateSendStatus(r, e)) {
					result.completeExceptionally(new RuntimeException("send fail ..."));
				}
			});
		});
		return result;
	}
	@Comment("用户建立连接后发送用户状态，好友在线状态消息")
	private CompletableFuture<Integer> sendStatusMessage(MessageResult messageResult, List<Integer> recivers) {
		CompletableFuture<Integer> result = new CompletableFuture<>();
		logger.finest("sendStatusMessage === " + jsonConvert.convertTo(messageResult) + ", recivers ==="
				+ jsonConvert.convertTo(recivers));
		recivers.forEach(reciver -> {
			wsmessageQueue.add(() -> {
				if (webSocketNode == null) {
					logger.log(Level.FINEST, "sendMap: " + JsonConvert.root().convertTo(messageResult));
					result.complete(0);
					return;
				}
				webSocketNode.sendMessage(messageResult, reciver).whenComplete((r, e) -> {
					logger.finest("sendStatusMessage websocket send status = " + r + ",execption =" + e + ",data==="
							+ jsonConvert.convertTo(messageResult));
					if (!validateSendStatus(r, e)) {
						result.completeExceptionally(new RuntimeException("send fail ..."));
					} else {
						result.complete(r);
					}
				});

			});
		});
		return result;
	}

	@Comment("校验消息发送状态")
	private boolean validateSendStatus(int status, Throwable error) {
		if (error != null) {
			return false;
		}
		if (status == WebSocket.RETCODE_SEND_ILLPACKET) {
			return false;
		}
		if (status == WebSocket.RETCODE_WSOCKET_CLOSED) {
			return false;
		}
		if (status == WebSocket.RETCODE_ILLEGALBUFFER) {
			return false;
		}
		if (status == WebSocket.RETCODE_SENDEXCEPTION) {
			return false;
		}
		if (status == WebSocket.RETCODE_ENGINE_NULL) {
			return false;
		}
		if (status == WebSocket.RETCODE_NODESERVICE_NULL) {
			return false;
		}
		if (status == WebSocket.RETCODE_GROUP_EMPTY) {
			return false;
		}
		if (status == WebSocket.RETCODE_WSOFFLINE) {
			return false;
		}
		if (status == WebSocket.RETCODE_DEAYSEND) {
			return false;
		}
		return true;
	}

	public FileService getFileService() {
		return fileService;
	}

	public void setFileService(FileService fileService) {
		this.fileService = fileService;
	}

	public WebSocketNode getWebSocketNode() {
		return webSocketNode;
	}

	public void setWebSocketNode(WebSocketNode webSocketNode) {
		this.webSocketNode = webSocketNode;
	}

	@Comment("数据源")
	public DataSource getSource() {
		return source;
	}

	public FriendService getFriendService() {
		return friendService;
	}

	public void setFriendService(FriendService friendService) {
		this.friendService = friendService;
	}

	@Comment("解密返回一个新对象数据")
	public synchronized MessageRecord decodeContent(MessageRecord record) {
		if (null == record) {
			return null;
		}
		// 如果开启就进行解密
		if (record.getEncrypt() == MessageRecord.ENCRYPT_ENABLE) {
			int targetid = record.getTargetid();
			short kindType = record.getMsgkind();
			PrivateKey privateKey = null;
			String content = record.getContent();
			if (kindType == MessageRecord.MSG_KIND_GROUP) {
				GroupDetail groupInfo = source.find(GroupDetail.class, targetid);
				privateKey = crypto.readPrivateKey(groupInfo.getEcdhprivkey());
				content = crypto.decrypt(record.getContent(), privateKey);
			}
			if (kindType == MessageRecord.MSG_KIND_PRIVATE) {
				UserInfo userInfo = source.find(UserInfo.class, targetid);
				String ecdhprivkey = userInfo.getEcdhprivkey();
				privateKey = crypto.readPrivateKey(ecdhprivkey);
				content = crypto.decrypt(record.getContent(), privateKey);
			}
			return record.copy(content);
		}
		return record;
	}

	public RetResult onMessage(UserInfo userInfo, MessageRecord bean) {
		logger.finer("onMessage .... bean=======>>>" + bean);
		
		RetResult retResult = RetResult.success();
		int targetId = bean.getTargetid();
		int formId = userInfo.getUserid();
		String messageid = formId + "-" + targetId + "-" + Utility.format36time(System.currentTimeMillis());
		
		if (bean.getMessageid() == null || bean.getMessageid().isEmpty()) {
			bean.setMessageid(messageid); // 此处由客户端生成 以免不好处理已读未读消息
		}
		bean.setCreatetime(System.currentTimeMillis());
		bean.setStatus(MessageRecord.UNREAD_STATUS);// 未读
		bean.setFromid(formId);
		bean.setOperatorUserId(formId);
		bean.setOperatorUserName(userInfo.getUsername());
		// 添加统一异常处理 是添加异步回调还是 提供双向链路给网关
		// 异常消息不处理
		// 消息校验
		MessageHandler handler = MessageHandler.fetchHandler(bean.getMsgtype());
		retResult = handler.validateMessage(this, bean).join();
		SendMsgResponse sendMsgResponse = new SendMsgResponse(bean.getMessageid(), bean.getTargetid(),
				bean.getMsgtype(), bean.getMsgkind(), bean.getCreatetime());
		retResult.setResult(sendMsgResponse);
		if (!retResult.isSuccess()) {
			return retResult;
		}
		// 消息处理
		CompletableFuture.supplyAsync(() -> handler.handler(this, bean).join()).whenComplete((t, ex) -> {
			if (ex != null) {
				logger.finest("handder error =====" + ex.getMessage());
			}
		});
                
                if (bean.getMsgtype() == 2017) {
                    bean.setContentObj(JsonConvert.root().convertFrom(NotifyContent.ReCallAllNotify.class, bean.getContent()));
                }
		// 解密后的数据
		insertQueue.add(bean.copy(jsonConvert.convertTo(bean.getContentObj())));// 保存入库
                logger.finest("clazz record aaaaaaaaaaaaaaaaaaaaaaaaaaa " + bean + " obj ===" + bean.getContentObj());
		multiTerminalMsg(userInfo, bean.copy(jsonConvert.convertTo(bean.getContentObj()))); // 广播
		return retResult;
	}
	
	@Comment("消息转换及其存储")
	@RestMapping(name = "onMessage", auth = true, comment = "消息转换及其存储")
	public RetResult onMessage(UserInfo userInfo,
				MessageRecord bean,
				@RestHeader(name = "CONNECTOR") String connector,
				@RestHeader(name = "LOCAL") String local) {
		logger.finer("onMessage .... bean=======>>>" + bean);
		bean.setClienttype(connector);
		bean.setLocal(local);
		return this.onMessage(userInfo, bean);
	}

	@Comment("多端同步")
	public void multiTerminalMsg(UserInfo user, MessageRecord record) {
		int msgtype = record.getMsgtype();
		MessageRecord notify =	SynMulTermalNotify.fromMessageRecord(record);
		if (filterMultilMsg.containsKey(msgtype)) {
			sendNotifyMessage(notify, Arrays.asList(user.getUserid()));
                        try {
                                // 保存相关记录
                               if(!user.isAppOnline()) {
                                   source.insert(notify);
                                   source.insert(notify.copyhis());//历史表
                               }
                        } catch (Exception e) {
                            e.printStackTrace();
                            logger.finest("error =====>>> " + e.getMessage());
                        }
                       
		} 
                // App不在线进行多端同步
                if("WEB".equals(record.getClienttype()) && !user.isAppOnline()) {
                        insertQueue.add(notify.copy(user.getUserid()));// 添加离线消息
                        logger.finest("save notify =======>>" + notify.toString());
                }
		
	}

	@RestMapping(name = "stickyCvsation", auth = true, comment = "会话置顶")
	public RetResult stickyCvsation(UserInfo userInfo, CvsationstickyRecord bean) {
		bean.setUserid(userInfo.getUserid());
		bean.setCreatetime(System.currentTimeMillis());
		if (bean.getOptype() == CvsationstickyRecord.OP_SAVE) {
			CvsationstickyRecord data = source.find(CvsationstickyRecord.class, bean.getCvsationstickyid());
			if (data == null) {
				insertQueue.add(bean);
			}
		} else if (bean.getOptype() == CvsationstickyRecord.OP_REMOVE) {
			deleteQueue.add(() -> {
				source.delete(CvsationstickyRecord.class, bean.getCvsationstickyid());
			});// 删除
		} else {
			return RetCodes.retResult(RetCodes.RET_PARAMS_ILLEGAL);
		}
		return RetResult.success();
	}

	@RestMapping(name = "queryStickyCvsations", auth = true, comment = "查询指定会话")
	public RetResult queryStickyCvsations(UserInfo userInfo) {
		return RetResult.success(source.queryList(CvsationstickyRecord.class,
				FilterNode.create("userid", FilterExpress.EQUAL, userInfo.getUserid())));
	}

	@Comment("根据开始时间计算所有表的创建时间")
	public Set<Long> fetchCreateTimes(long startTime, long endtime) {
		HashSet<Long> createTimes = new HashSet<>();
		LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(startTime), ZoneOffset.ofHours(8));
		LocalDateTime endTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(endtime), ZoneOffset.ofHours(8));
		do {
			createTimes.add(time.toInstant(ZoneOffset.UTC).toEpochMilli());
			time = time.plusDays(1);
		} while (endTime.isAfter(time));
		return createTimes;
	}

	@RestMapping(name = "enter", auth = true, comment = "进入系统")
	public RetResult enter(UserInfo userInfo, @RestHeader(name = "WS-SncpAddress") InetSocketAddress sncpAddress,
			@RestHeader(name = "CONNECTOR") String connector, @RestHeader(name = "JSESSIONID") String jsessionid) {
		sendOnlineStatus(userInfo.getUserid(), FriendShip.ONLINE_STATUS_ONLINE);
		sendOnlineFriends(userInfo.getUserid());// 发送在线用户信息
		// 新用户发送新通知
		if (UserInfo.USER_CYCLE_NEW == userInfo.getUsercycle()) {
			try {
				sendSysNotice(userInfo);
			}catch (Exception e) {
				e.printStackTrace();
			}
		}

		// 发送H5 断开连接离线消息
		if ("WEB".equalsIgnoreCase(connector)) {
			// h5SessionMapper.remove(jsessionid);
                        // 处理老用户  App删除的公告
                } else {
                    // 关闭公告
                    try {
                            sendCloseNotice(userInfo);
			} catch (Exception e) {
                            e.printStackTrace();
			}
                }
		//上线更新终端
		updateUserOnlineTerimal(userInfo, connector, true);
		return RetResult.success();
	}

        /**
         * 发送关闭公告
         */
        private void sendCloseNotice(UserInfo userInfo) {
              Flipper  flipper = new Flipper();
                flipper.sort("createtime desc");
            FilterNode node = FilterNode
                        .create("status", FilterExpress.EQUAL, SystemNotice.CLOSE_STATUS)
                        .and("userid", FilterExpress.EQUAL, userInfo.getUserid());
             List<SystemUserNotice> list =  source.queryList(SystemUserNotice.class, flipper, node);
             
             if (!list.isEmpty()) {
              
               /**
                * 获取最新的一条
                */
                List<SystemUserNotice> rows =  source.queryList(SystemUserNotice.class,flipper, FilterNode
					.create("userid", FilterExpress.EQUAL, userInfo.getUserid())
					.and("status", FilterExpress.EQUAL, SystemNotice.OPEN_STATUS));
                SystemUserNotice bean = list.get(0);
                SysNoticeMessage message = new SysNoticeMessage();
                message.setNotice(bean);
                // 关闭状态
                message.setStatus(SysNoticeMessage.CLOSE_STATUS);
                if (!rows.isEmpty()) {
                     message = message.copyLastNotice(rows.get(0));
                }
                MessageRecord messageRecord = new MessageRecord();
                /**
                 * 查询所有的系统用户 发送相关消息
                 */
                messageRecord.setContent(JsonConvert.root().convertTo(message));
                messageRecord.setMsgtype(SysNoticeMessage.TYPE);
                messageRecord.setFromid(UserInfo.USER_MSG_SYSTEM.getUserid());
                messageRecord.setCreatetime(System.currentTimeMillis());
                messageRecord.setMsgkind(MessageRecord.MSG_KIND_SYSTEM);
                messageRecord.setMessageid(UserInfo.USER_MSG_SYSTEM.getUserid() + "-" + Utility.format36time(System.currentTimeMillis()));
                messageRecord.setTargetid(userInfo.getUserid());
                sendNotifyMessageOnly(messageRecord, Arrays.asList(userInfo.getUserid()));
                 // 删除
                source.delete(list);
             }
        }
	/**
	 * 更新当前用户在线终端类型
	 * @param userInfo
	 * @param connector
	 */
	
	private void updateUserOnlineTerimal(UserInfo userInfo, String connector, boolean online) {
		int onlineTerminal = userInfo.getOnlineterminal();
		logger.finest("userid = " + userInfo.getUserid() + ", connector = " + connector +", online=" + online+", onlineTerminal =" + onlineTerminal);
		if("Android".equals(connector)
					|| connector.equalsIgnoreCase("IOS")) {
			if (online) {
				onlineTerminal = onlineTerminal | 1;
				// web 端不在线
				if((onlineTerminal & 2) != 2) {
					sendDisableH5MsgOnly(userInfo.getUserid());
				} else {
					sendOnlineH5MsgOnly(userInfo.getUserid());
				}
			} else {
				onlineTerminal = onlineTerminal & 2;
			}
		}
		
		if("WEB".equalsIgnoreCase(connector)) {
			if (online) {
				onlineTerminal = onlineTerminal | 2;//与第二位求 或 |
				sendOnlineH5MsgOnly(userInfo.getUserid());
			} else {
				onlineTerminal = onlineTerminal & 1;// 与第一位求 &
                                sendDisableH5MsgOnly(userInfo.getUserid());// H5下线
			}
		}
		// 更新在线终端
		userInfo.setOnlineterminal(onlineTerminal);
		source.updateColumn(UserDetail.class, userInfo.getUserid(), ColumnValue.mov("onlineterminal", onlineTerminal));
	    source.updateColumn(UserInfo.class, userInfo.getUserid(), ColumnValue.mov("onlineterminal", onlineTerminal));
	}
	@Comment("发送新用户系统通知")
	private void sendSysNotice(UserInfo userInfo) {
		List<SystemNotice> list = source.queryList(SystemNotice.class,
				FilterNode.create("noticetarget", FilterExpress.EQUAL, SystemNotice.USER_CYCLE_NEW)
							.and("status", FilterExpress.EQUAL, SystemNotice.OPEN_STATUS));
		if (!list.isEmpty()) {
			list.forEach(bean -> {
				SysNoticeMessage message = new SysNoticeMessage();
				message.setNotice(bean);
				MessageRecord messageRecord = new MessageRecord();
				/**
				 * 查询所有的系统用户 发送相关消息
				 */
				messageRecord.setContent(JsonConvert.root().convertTo(message));
				messageRecord.setMsgtype(SysNoticeMessage.TYPE);
				messageRecord.setFromid(UserInfo.USER_MSG_SYSTEM.getUserid());
				messageRecord.setCreatetime(System.currentTimeMillis());
				messageRecord.setMsgkind(MessageRecord.MSG_KIND_SYSTEM);
				messageRecord.setMessageid(
						UserInfo.USER_MSG_SYSTEM.getUserid() + "-" + Utility.format36time(System.currentTimeMillis()));
				insertQueue.add(messageRecord);
				insertQueue.add(bean.copySystemUserNotice(userInfo.getUserid()));//拷贝对象查询使用
				sendNotifyMessageWithSave(messageRecord, Arrays.asList(userInfo.getUserid()));
			});
		}

	}

	/**
	 * todo 联调 离线消息 时候需要同客户端确认处理方式 ： 1.消息推过去 2.客户端直接去拉消息
	 *
	 * @param userInfo
	 * @return
	 */
	@RestMapping(name = "countOfflineMessage", auth = true, comment = "离线消息统计")
	public RetResult countOfflineMessage(UserInfo userInfo) {
               
		// 查询群信息
		CompletableFuture<Set<OfflineMsgRecord>> groupMsgFuture = source
				.querySetAsync(OfflineMsgRecord.class, SelectColumn.includes("msgkind", "targetid"),
						FilterNode
								.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
								.and("msgkind", FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_GROUP)
                                                                .and("msgtype", FilterExpress.NOTEQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE))//不统计多端同步
				.thenApply((dataSet) -> {
					dataSet.forEach(record -> {
                                                int targetid = record.getTargetid();
                                                FilterNode multermalNode = FilterNode.create("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
						FilterNode node = FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
								.and("msgkind", FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_GROUP)
								.and("targetid", FilterExpress.EQUAL, record.getTargetid());
                                                                
						FilterNode countNode = FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
								.and("msgkind", FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_GROUP)
								 // 统计数据消息范围
								.and(FilterNode.create("msgtype", FilterExpress.GREATERTHANOREQUALTO, 1001).and("msgtype", FilterExpress.LESSTHANOREQUALTO, 1013)
                                                                            //.or("msgtype", FilterExpress.EQUAL, OBJ_SYNMULTERNAL_MESSAGE)//多端同步统计添加
                                                                ) 
								.and("targetid", FilterExpress.EQUAL, record.getTargetid());
						Number num = source.getNumberResult(OfflineMsgRecord.class, FilterFunc.COUNT, 0, "messageid",
								countNode);
						
                                                FilterNode lastMultermalNode = FilterNode.create("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                                {
                                                            lastMultermalNode.and("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("fromid",FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("targetid",FilterExpress.EQUAL, record.getTargetid())
                                                            .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                                            lastMultermalNode.or(countNode);
                                                }
                                                
                                                // 最新一条消息
                                                Flipper lastfipper = new Flipper(1, "createtime desc");
						//List<OfflineMsgRecord> lastlist = source.queryList(OfflineMsgRecord.class, lastfipper, countNode);
                                                List<OfflineMsgRecord> lastlist = source.queryList(OfflineMsgRecord.class, lastfipper, lastMultermalNode);
                                                if (!lastlist.isEmpty()) {
                                                    OfflineMsgRecord lastRecord =  lastlist.get(0);
                                                    record.setMessageid(lastRecord.getMessageid());
                                                    record.setContent(lastRecord.getContent());
                                                    record.setMsgkind(lastRecord.getMsgkind());
                                                    record.setFromid(lastRecord.getFromid());
                                                    record.setTargetid(lastRecord.getTargetid());
                                                    record.setMsgtype(lastRecord.getMsgtype());
                                                    record.setCreatetime(lastRecord.getCreatetime());
                                                }
                                                
                                                //自己给群里发的消息
                                                multermalNode.and("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("fromid",FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                                multermalNode.or(node);
                                                
                                                // 最后一条时间
						Flipper flipper = new Flipper(1, "createtime desc");
						List<OfflineMsgRecord> list = source.queryList(OfflineMsgRecord.class, flipper, node);
						if (list != null && !list.isEmpty()) {
							OfflineMsgRecord offlineMsgRecord = list.get(0);
							record.setMaxtime(offlineMsgRecord.getCreatetime());
						}
						flipper = new Flipper(1, "createtime asc");
						List<OfflineMsgRecord>  setList = source.queryList(OfflineMsgRecord.class, flipper, node);
						// 设置最小时间
						if (setList != null && !setList.isEmpty()) {
							OfflineMsgRecord offlineMsgRecord = setList.get(0);
							record.setMintime(offlineMsgRecord.getCreatetime());
						}
						record.setTotal(num.intValue());
					});
					return dataSet;
				});

		CompletableFuture<Set<OfflineMsgRecord>> privateMsgFuture = source
				.querySetAsync(OfflineMsgRecord.class, SelectColumn.includes("msgkind", "fromid", "targetid"),
						FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid()).and("msgkind",
								FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_PRIVATE)
                                                            .and("msgtype", FilterExpress.NOTEQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE))//不统计多端同步
				.thenApply((dataSet) -> {
                                         
					dataSet.forEach((OfflineMsgRecord record) -> {
                                                FilterNode multermalNode = FilterNode.create("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                             
						FilterNode node = FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
								.and("msgkind", FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_PRIVATE)
								.and("targetid", FilterExpress.EQUAL, record.getTargetid());
						
						
						//统计条件
						FilterNode countNode = FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
								.and("msgkind", FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_PRIVATE)
								.and(FilterNode.create("msgtype", FilterExpress.GREATERTHANOREQUALTO, 1001)
                                                                            .and("msgtype", FilterExpress.LESSTHANOREQUALTO, 1013)
                                                                 //.or("msgtype", FilterExpress.EQUAL, OBJ_SYNMULTERNAL_MESSAGE)//多端同步统计添加)
                                                                ) 
                                                        
								.and("targetid", FilterExpress.EQUAL, record.getTargetid());
                                                
						Number num = source.getNumberResult(OfflineMsgRecord.class, FilterFunc.COUNT, 0, "messageid", countNode);
                                                
						FilterNode lastMultermalNode = FilterNode.create("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                                {
                                                            lastMultermalNode.and("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("fromid",FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("targetid",FilterExpress.EQUAL, record.getTargetid())
                                                            .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                                            lastMultermalNode.or(countNode);
                                                }
                                                
						Flipper lastfipper = new Flipper(1, "createtime desc");
						// List<OfflineMsgRecord> lastlist = source.queryList(OfflineMsgRecord.class, lastfipper, countNode);
                                                List<OfflineMsgRecord> lastlist = source.queryList(OfflineMsgRecord.class, lastfipper, lastMultermalNode);
                                                if (!lastlist.isEmpty()) {
                                                   OfflineMsgRecord lastRecord =  lastlist.get(0);
                                                    record.setMessageid(lastRecord.getMessageid());
                                                    record.setContent(lastRecord.getContent());
                                                    record.setMsgkind(lastRecord.getMsgkind());
                                                    record.setFromid(lastRecord.getFromid());
                                                    record.setTargetid(lastRecord.getTargetid());
                                                    record.setMsgtype(lastRecord.getMsgtype());
                                                    record.setCreatetime(lastRecord.getCreatetime());
                                                }
                                                
                                                multermalNode.and("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
                                                            //.and("fromid",FilterExpress.EQUAL, userInfo.getUserid())
                                                            .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
                                                
                                                multermalNode.or(node);
						Flipper flipper = new Flipper(1, "createtime desc");
						List<OfflineMsgRecord> list = source.queryList(OfflineMsgRecord.class, flipper, multermalNode);
						if (list != null && !list.isEmpty()) {
							OfflineMsgRecord offlineMsgRecord = list.get(0);
							record.setMaxtime(offlineMsgRecord.getCreatetime());
						}
						flipper = new Flipper(1, "createtime asc");
						List<OfflineMsgRecord>  setList = source.queryList(OfflineMsgRecord.class, flipper, multermalNode);
						// 设置最小时间
						if (setList != null && !setList.isEmpty()) {
							OfflineMsgRecord offlineMsgRecord = setList.get(0);
							record.setMintime(offlineMsgRecord.getCreatetime());
						}
						record.setTotal(num.intValue());
					});
					return dataSet;
				});

		// 系统公告查询
		CompletableFuture<Set<OfflineMsgRecord>> sysMsgFuture = source
				.querySetAsync(OfflineMsgRecord.class, SelectColumn.includes("msgkind", "fromid", "targetid"),
						FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid()).and("msgkind",
								FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_SYSTEM))
				.thenApply((dataSet) -> {
					dataSet.forEach((OfflineMsgRecord record) -> {
						FilterNode node = FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
								.and("msgkind", FilterExpress.EQUAL, OfflineMsgRecord.MSG_KIND_SYSTEM)
								.and("msgtype", FilterExpress.EQUAL, SysNoticeMessage.TYPE)
								.and("targetid", FilterExpress.EQUAL, record.getTargetid());
						Number num = source.getNumberResult(OfflineMsgRecord.class, FilterFunc.COUNT, 0, "messageid",
								node);
						Flipper flipper = new Flipper(1, "createtime desc");

						List<OfflineMsgRecord> list = source.queryList(OfflineMsgRecord.class, flipper, node);
						if (list != null && !list.isEmpty()) {
							OfflineMsgRecord offlineMsgRecord = list.get(0);
							record.setMessageid(offlineMsgRecord.getMessageid());
							record.setContent(offlineMsgRecord.getContent());
							record.setMsgkind(offlineMsgRecord.getMsgkind());
							record.setFromid(offlineMsgRecord.getFromid());
							record.setTargetid(offlineMsgRecord.getTargetid());
							record.setMsgtype(offlineMsgRecord.getMsgtype());
							record.setCreatetime(offlineMsgRecord.getCreatetime());
							record.setMaxtime(offlineMsgRecord.getCreatetime());
						}
						flipper = new Flipper(1, "createtime asc");
						list = source.queryList(OfflineMsgRecord.class, flipper, node);
						// 设置最小时间
						if (list != null && !list.isEmpty()) {
							OfflineMsgRecord offlineMsgRecord = list.get(0);
							record.setMintime(offlineMsgRecord.getCreatetime());
						}
						record.setTotal(num.intValue());
					});
					return dataSet;
				});

		CompletableFuture completableFuture = groupMsgFuture.thenCombine(privateMsgFuture,
				new BiFunction<Set<OfflineMsgRecord>, Set<OfflineMsgRecord>, Object>() {
					@Override
					public Object apply(Set<OfflineMsgRecord> arg0, Set<OfflineMsgRecord> arg1) {
//						source.delete(OfflineMsgRecord.class,
//								FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid()));
						arg0.addAll(arg1);
						return arg0;
					}
				});

		CompletableFuture<Set<OfflineMsgRecord>> resultFeature = sysMsgFuture.thenCombine(completableFuture,
				new BiFunction<Set<OfflineMsgRecord>, Set<OfflineMsgRecord>, Object>() {
					@Override
					public Object apply(Set<OfflineMsgRecord> arg0, Set<OfflineMsgRecord> arg1) {
//						source.delete(OfflineMsgRecord.class,
//								FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid()));
						arg0.addAll(arg1);
						return arg0;
					}
				});
                Set<OfflineMsgRecord> setrecords =  resultFeature.join();
                RetResult rs = RetResult.success(resultFeature.join());
                rs.attach("multitermal", countMultiTerminalMsg(userInfo, setrecords));//多端消息同步
                source.delete(OfflineMsgRecord.class, FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid()));
		return rs;
	}
        

        private Collection<OfflineMsgRecord> countMultiTerminalMsg(UserInfo userInfo, Set<OfflineMsgRecord> setrecords) {
                Set<OfflineMsgRecord> collections = source.querySet(OfflineMsgRecord.class, 
                                                        FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
                                                        .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE));
                Set<Integer> userSet = new HashSet();
                Map<Integer, OfflineMsgRecord> sendMap = new HashMap();
                if (setrecords != null && !setrecords.isEmpty()) {
                    setrecords.forEach(record -> {
                        userSet.add(record.getFromid());
                    });
                    
                }
                //当前集合处理
                if (!collections.isEmpty()) {
                  
                    collections.forEach(record -> {
                        Integer userid = record.getTargetid();
                        if (!sendMap.containsKey(userid) && !userSet.contains(userid)) {
                            sendMap.put(userid, record);
                        }
                    });
                    
                    sendMap.values().forEach(record -> {
                        FilterNode node = FilterNode.create("receiverid", FilterExpress.EQUAL, userInfo.getUserid())
                                                                .and("targetid", FilterExpress.EQUAL, record.getTargetid());
                       Flipper flipper = new Flipper(1, "createtime asc");
                       List<OfflineMsgRecord> list = source.queryList(OfflineMsgRecord.class, flipper, node);
                        // 设置最小时间
                       if (list != null && !list.isEmpty()) {
                               OfflineMsgRecord offlineMsgRecord = list.get(0);
                               record.setMintime(offlineMsgRecord.getCreatetime());
                       }
                       // 最大时间
                       flipper = new Flipper(1, "createtime desc");
                       List<OfflineMsgRecord> datSet = source.queryList(OfflineMsgRecord.class, flipper, node);
                        // 设置最小时间
                       if (datSet != null && !datSet.isEmpty()) {
                               OfflineMsgRecord offlineMsgRecord = datSet.get(0);
                               record.setMaxtime(offlineMsgRecord.getCreatetime());
                       }
                    });
                }        
                return sendMap.values();
        }
        
	/**
	 * @param userInfo
	 * @return
	 */
	@RestMapping(name = "loadOfflineMessage", auth = true, comment = "查询离线消息")
	public RetResult loadOfflineMessage(UserInfo userInfo, Flipper flipper, OfflineMsgRecord bean) {
		long maxtime = bean.getCreatetime();
		long mintime = bean.getMintime();
                // 多端消息同步类型
                FilterNode multermalNode = FilterNode.create("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE);
		if (bean.getCreatetime() == 0l) {
			return RetCodes.retResult(RetCodes.RET_PARAMS_ILLEGAL);
		}
		if (flipper == null) {
			flipper = new Flipper(65535); // 客户端要求 一次查询完
		}
		flipper.sort("createtime asc");
		short msgkind = bean.getMsgkind();
		// 群消息
		Sheet<MessageRecord> sheet = null;
		// 群聊
		if (msgkind == OfflineMsgRecord.MSG_KIND_GROUP) {
			if (bean.getTargetid() == 0) {
				return RetCodes.retResult(RetCodes.RET_PARAMS_ILLEGAL);
			}
                        FilterNode offlineNode =FilterNode.create("createtime", FilterExpress.LESSTHANOREQUALTO, maxtime)
							.and("createtime", FilterExpress.GREATERTHANOREQUALTO, mintime)
							// 消息范围
							.and(FilterNode.create("msgtype", FilterExpress.GREATERTHANOREQUALTO 
                                                                        , NotifyContent.OBJ_UPDATE_USERNICKNAME_NTF)
									.or("msgtype", FilterExpress.LESSTHAN, NotifyContent.OBJ_READ_NTF))
                                                                        .and("targetid", FilterExpress.EQUAL, bean.getTargetid())
                                                                        .and("fromid", FilterExpress.NOTEQUAL, userInfo.getUserid());//查询群成员发送的
                                
                        multermalNode.and("createtime", FilterExpress.LESSTHANOREQUALTO, maxtime)
                                     .and("createtime", FilterExpress.GREATERTHANOREQUALTO, mintime)
                                     .and("fromid", FilterExpress.EQUAL, userInfo.getUserid())
                                     .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE)
                                     .and("targetid", FilterExpress.EQUAL, bean.getTargetid());
                        
                        offlineNode.or(multermalNode);
                        sheet = source.querySheet(MessageRecord.class, flipper, offlineNode);
		}
		// 私聊
		if (msgkind == OfflineMsgRecord.MSG_KIND_PRIVATE) {
                    
                        FilterNode offlineNode = FilterNode.create("createtime", FilterExpress.LESSTHANOREQUALTO, maxtime)
							.and("createtime", FilterExpress.GREATERTHANOREQUALTO, mintime)
							.and("msgtype", FilterExpress.NOTEQUAL,NotifyContent.OBJ_SYNMULTERNAL_MESSAGE)//不是多端同步
							.and("fromid", FilterExpress.EQUAL, bean.getFromid())
							.and("targetid", FilterExpress.EQUAL, userInfo.getUserid());
                        //多端同步消息类型
                        multermalNode.and("createtime", FilterExpress.LESSTHANOREQUALTO, maxtime)
                                     .and("createtime", FilterExpress.GREATERTHANOREQUALTO, mintime)
                                    .and("msgtype", FilterExpress.EQUAL, NotifyContent.OBJ_SYNMULTERNAL_MESSAGE)
                                    .and("fromid", FilterExpress.EQUAL, userInfo.getUserid())
                                    .and("targetid", FilterExpress.EQUAL, bean.getFromid());
                        offlineNode.or(multermalNode);
			sheet = source.querySheet(MessageRecord.class, flipper,offlineNode);
		}

		// 系统消息
		if (msgkind == OfflineMsgRecord.MSG_KIND_SYSTEM) {
			sheet = source.querySheet(MessageRecord.class, flipper,
					FilterNode.create("createtime", FilterExpress.LESSTHANOREQUALTO, maxtime)
							.and("createtime", FilterExpress.GREATERTHANOREQUALTO, mintime)
							.and(FilterNode.create("msgtype", FilterExpress.EQUAL, BasicContent.OBJ_SYSNOTICE_MESSAGE))
							.and("targetid", FilterExpress.EQUAL, userInfo.getUserid()));
		}
                
                if (sheet.getRows() != null && !sheet.getRows().isEmpty()) {
                    sheet.getRows().forEach(record -> {
                        short  msgtype = record.getMsgtype();
                        if (msgtype == NotifyContent.OBJ_SYNMULTERNAL_MESSAGE) {
                            record.setClienttype("WEB");
                        }
                    });
                }
                
		return RetResult.success(sheet);
	}

        
        
        
	@RestMapping(name = "leave", auth = true, comment = "退出系统")
	public RetResult leave(UserInfo userInfo, @RestHeader(name = "WS-SncpAddress") InetSocketAddress sncpAddress,
			@RestHeader(name = "CONNECTOR") String connector, @RestHeader(name = "JSESSIONID") String jsessionid) {
		sendOnlineStatus(userInfo.getUserid(), FriendShip.ONLINE_STATUS_OFFLINE);// 发送离线状态给好友
		// 新用户发送新通知
		if (UserInfo.USER_CYCLE_NEW == userInfo.getUsercycle() || userInfo.getUsercycle() == 0) {
			source.updateColumn(UserDetail.class, userInfo.getUserid(), "usercycle", UserInfo.USER_CYCLE_OLD);
			source.updateColumn(UserInfo.class, userInfo.getUserid(), "usercycle", UserInfo.USER_CYCLE_OLD);
		}
		// 发送H5 断开连接离线消息
                //		if ("WEB".equalsIgnoreCase(connector)) {
                //			long now = System.currentTimeMillis();
                //			h5SessionMapper.put(jsessionid, userInfo.getUserid() + "-" + Utility.format36time(now));
                //		}
		//上线更新终端
                if (!"WEB".equalsIgnoreCase(connector)) {
                    connector = "Android";
                }
                logger.finest("App xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx========================" + connector);
		updateUserOnlineTerimal(userInfo, connector, false);
		return RetResult.success();
	}

	@Comment("发送 H5下线通知")
	private void sendDisableH5Msg(String jsessionId, int userid) {
		UserInfo userInfo = source.find(UserInfo.class, userid);
		logger.finest("发送 H5下线通知  jsessionId ====" + jsessionId);
		if (userInfo == null) return;
		logger.finest("发送 H5下线通知  userid ====" + userid);
		// 发送H5 离线消息
		MessageRecord message = new MessageRecord();
		long now = System.currentTimeMillis();
		message.setCreatetime(now);
		message.setMessageid(userInfo.getUserid() + "-" + userInfo.getUserid() + Utility.format36time(now));
		message.setFromid(userInfo.getUserid());
		message.setTargetid(userInfo.getUserid());
		message.setMsgkind(MessageRecord.MSG_KIND_SYSTEM);
		message.setMsgtype(NotifyContent.OBJ_DISABLEAPP_H5);
		NotifyContent.DisableH5Status notify = new NotifyContent.DisableH5Status();
                // 更新web 在线状态
                if (userInfo.isWebOnline()) {
                    // App online 1
                    userInfo.setOnlineterminal(userInfo.getOnlineterminal() & 1);
                    source.updateColumn(UserDetail.class, userInfo.getUserid(), ColumnValue.mov("onlineterminal", userInfo.getOnlineterminal()));
                    source.updateColumn(UserInfo.class, userInfo.getUserid(), ColumnValue.mov("onlineterminal", userInfo.getOnlineterminal()));
                }
		notify.setExtra(userInfo);
		message.setContent(JsonConvert.root().convertTo(notify));
		sendMessage(message);// 发送H5 断开连接离线消息
		userService.logout(jsessionId);// 注销登录
		logger.finest("发送 H5下线通知  logout ====" + jsessionId);
	}

	/**
	 * app 登录发送 web在线状态
	 * @param userid
	 */
	
	private void sendDisableH5MsgOnly(int userid) {
		UserInfo userInfo = source.find(UserInfo.class, userid);
		if (userInfo == null) return;
		logger.finest("发送 H5下线通知  userid ====" + userid);
		// 发送H5 离线消息
		MessageRecord message = new MessageRecord();
		long now = System.currentTimeMillis();
		message.setCreatetime(now);
		message.setMessageid(userInfo.getUserid() + "-" + userInfo.getUserid() + Utility.format36time(now));
		message.setFromid(userInfo.getUserid());
		message.setTargetid(userInfo.getUserid());
		message.setMsgkind(MessageRecord.MSG_KIND_SYSTEM);
		message.setMsgtype(NotifyContent.OBJ_DISABLEAPP_H5);
		NotifyContent.DisableH5Status notify = new NotifyContent.DisableH5Status();
		notify.setExtra(userInfo);
		message.setContent(JsonConvert.root().convertTo(notify));
		sendMessage(message);// 发送H5 断开连接离线消息
	}
	//发送H5 在线通知
	private void sendOnlineH5MsgOnly(int userid) {
		UserInfo userInfo = source.find(UserInfo.class, userid);
		if (userInfo == null) return;
		logger.finest("发送 H5下线通知  userid ====" + userid);
		// 发送H5 离线消息
		MessageRecord message = new MessageRecord();
		long now = System.currentTimeMillis();
		message.setCreatetime(now);
		message.setMessageid(userInfo.getUserid() + "-" + userInfo.getUserid() + Utility.format36time(now));
		message.setFromid(userInfo.getUserid());
		message.setTargetid(userInfo.getUserid());
		message.setMsgkind(MessageRecord.MSG_KIND_SYSTEM);
		message.setMsgtype(NotifyContent.OBJ_ONLINE_APP_H5);
		NotifyContent.OnlineH5Status notify = new NotifyContent.OnlineH5Status();
		notify.setExtra(userInfo);
		message.setContent(JsonConvert.root().convertTo(notify));
		sendMessage(message);// 发送H5 在线通知
	}
	
	@Comment("发送用户在线状态  不记录到数据库")
	private void sendOnlineStatus(int userid, short onlinestatus) {
		Set<Integer> receiveids = source.queryColumnSet("friendid", FriendShip.class,
				FilterNode.create("userid", FilterExpress.EQUAL, userid));
		List<Integer> ids = new ArrayList<>(receiveids);
		NotifyContent.UserOnlineStatusNotify contentObj = new NotifyContent.UserOnlineStatusNotify();
		contentObj.setUserid(userid);
		contentObj.setOnlinestatus(onlinestatus);
		MessageRecord record = new MessageRecord();
		record.setFromid(userid);
		record.setContent(jsonConvert.convertTo(contentObj));
		record.setMsgtype(NotifyContent.UserOnlineStatusNotify.TYPE);// 用户在线状态
		sendStatusMessage(MessageResult.buildRecord(record), ids); // 发送上线状态给好友
	}

	@Comment("发送上线前在线好友的id列表  不记录到数据库")
	private void sendOnlineFriends(int userid) {
		// todo 加载群在线用户
		// 加载好友在线用户
		Set<String> onlineUserSet = webSocketNode.getUserSet().join();
		if (onlineUserSet == null || onlineUserSet.size() == 0) {
			return; // 在线用户不存在直接返回
		}
		Set<Integer> onlineUsers = onlineUserSet.stream().map(x -> Integer.valueOf(x)).collect(Collectors.toSet());
		Set<Integer> friendids = source.queryColumnSet("friendid", FriendShip.class,
				FilterNode.create("userid", FilterExpress.EQUAL, userid));
		if (friendids != null) {
			HashSet<Integer> onlinefriends = new HashSet(onlineUsers);
			onlinefriends.retainAll(friendids);
			NotifyContent.OnlineUserListNotify notify = new NotifyContent.OnlineUserListNotify();
			notify.setUserids(onlinefriends);
			notify.setObjType(NotifyContent.OnlineUserListNotify.FRIEND_LIST_ONLINE);
			MessageRecord record = new MessageRecord();
			record.setFromid(userid);
			record.setContent(jsonConvert.convertTo(notify));
			record.setMsgtype(NotifyContent.OnlineUserListNotify.TYPE);// 用户在线状态
			record.setTargetid(userid);
			sendStatusMessage(MessageResult.buildRecord(record), Arrays.asList(userid));// 发送在线列表给当前用户
		}

	}

	/**
	 * 获取当前用户信息
	 * @param userid
	 * @return
	 */
	private boolean appOnline(int userid) {
		UserInfo userinfo = source.find(UserInfo.class, userid);
		if (userinfo == null) return true;
		return userinfo.isAppOnline();
	}
        
        /**
	 * 获取当前用户信息
	 * @param userid
	 * @return
	 */
	private boolean appOnlineOss(int userid) {
		UserDetail userinfo = source.find(UserDetail.class, userid);
		if (userinfo == null) return true;
		return userinfo.isAppOnline();
	}
	public static void main(String[] args) throws Exception {
		MessageService service = Application.singleton(MessageService.class);
		UserInfo userInfo = UserInfo.USER_SYSTEM;
		userInfo.setUserid(304491669);
        //service.loadOfflineMessage(userInfo);
//        System.out.println("======" + service.countOfflineMessage(userInfo));
        OfflineMsgRecord bean = new OfflineMsgRecord();
        bean.setFromid(304491669);
        bean.setTargetid(102078943);
        bean.setMsgkind(Short.valueOf("30"));
        bean.setCreatetime(1587449155845l);
        bean.setMintime(  1587449141613l);
        System.out.println("data === " + service.loadOfflineMessage(userInfo, null, bean));
        Thread.sleep(1000 * 20);
                
//		OfflineMsgRecord record = new OfflineMsgRecord();
//		record.setTargetid(100000041);
//		record.setCreatetime(1578555835978l);
//		System.out.println(service.loadOfflineMessage(userInfo, null, record));
//		Thread.sleep(10000);
//        MessageRecord record = JsonConvert.root().convertFrom(MessageRecord.class, "{\"content\":\"{\\\"targetPic\\\":\\\"m1hq4y.jpg\\\",\\\"targettype\\\":\\\"10\\\",\\\"targetname\\\":\\\"哈哈哈\\\",\\\"targetid\\\":\\\"300000014\\\"}\",\"createtime\":1578043616812,\"fromid\":300000014,\"messageid\":\"300000014-300000042-k4xyfa3b\",\"msgkind\":20,\"msgtype\":2012,\"status\":10,\"targetid\":300000042}");
//        System.out.println(record.getContent());
//        BasicContent.ShareCardNotify o = JsonConvert.root().convertFrom(BasicContent.ShareCardNotify.class, "{\"targetPic\":\"m1hq4y.jpg\",\"targettype\":\"10\",\"targetname\":\"哈哈哈\",\"targetid\":\"300000014\"}");
//        UserInfo userInfo = new UserInfo();
//        userInfo.setUserid(300000014);
//        service.onMessage(userInfo, record);
//        System.out.println("record" + record.getContent());
//        System.out.println("record" + record);
	}
}
